package com.xxx.service;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.concurrent.*;

/**
 * 原生实现rocketmq事务消息①
 */
@Slf4j
@Service
public class TransactionalProducerService {

    @Value("${rocketmq.name-server}")
    private String nameServer;
    @Value("${rocketmq.producer.group}")
    private String group;

    private TransactionMQProducer producer;

    // 系统启动时初始化producer
    @PostConstruct
    public void init() throws MQClientException {
        // 1.创建事务消息生产者
        producer = new TransactionMQProducer(group);
        producer.setNamesrvAddr(nameServer);

        // 2.设置线程池用于执行本地事务
        ExecutorService executorService = new ThreadPoolExecutor(
                2,
                5,
                100,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue(1000),
                new ThreadPoolExecutor.CallerRunsPolicy()
        );
        producer.setExecutorService(executorService);

        // 3.注册事务监听器，需要实现2个接口：executeLocalTransaction（执行本地事务）、checkLocalTransaction（回查本地事务执行结果）
        producer.setTransactionListener(new TransactionListener() {
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                String orderId = new String(msg.getBody());
                System.out.println("【事务开始】执行本地事务，订单ID：" + orderId);

                try {
                    boolean success = execLocalTransaction(orderId);  // 执行本地事务
                    if (success) {
                        return LocalTransactionState.COMMIT_MESSAGE;  // 本地事务执行成功，返回commit，告诉RocketMQ可以投递消息
                    } else {
                        return LocalTransactionState.ROLLBACK_MESSAGE;  // 本地事务执行失败，返回rollback，告诉RocketMQ删除消息
                    }
                } catch (Exception e) {
                    return LocalTransactionState.UNKNOW;   // 发生一次，返回unknown，告诉RocketMQ稍后进行回查
                }
            }

            // 提供给RocketMQ的回查方法
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                String orderId = new String(messageExt.getBody());
                System.out.println("【事务回查】检查本地事务状态，订单ID：" + orderId);
                // 查询本地事务结果
                boolean isCommitted = checkDatabaseStatus(orderId);
                return isCommitted ? LocalTransactionState.COMMIT_MESSAGE : LocalTransactionState.ROLLBACK_MESSAGE;
            }

            // 模拟本地事务操作（数据库更新）
            private boolean execLocalTransaction(String orderId) {
                // 模拟业务逻辑，比如插入订单
                return !orderId.contains("fail"); // 成功或失败
            }

            private boolean checkDatabaseStatus(String orderId) {
                // 模拟回查数据库状态
                return true;
            }
        });

        // 4.启动事务消息生产者
        producer.start();
        log.info("事务消息生产者已启动");
    }

    /**
     * 发送事务消息
     * @param topic
     * @param msg
     * @throws MQClientException
     */
    public void sendTransactionalMessage(String topic, String msg) throws MQClientException {
        Message message = new Message(topic, "OrderTag", msg.getBytes());
        TransactionSendResult sendResult = producer.sendMessageInTransaction(message, null);
        log.info("事务消息发送结果: " + sendResult);
    }

    // 关闭producer
    public void shutdown() {
        if (producer != null) {
            producer.shutdown();
        }
    }
}
